﻿using Microsoft.Extensions.Logging;
using Nest;
using System.Reflection;
using System.Text.Json;

namespace In66.Net.Elasticsearch
{
    #region Repository
    public class ElasticsearchRepository<T> : IElasticsearchRepository<T> where T : ElasticsearchEntity
    {
        private readonly IElasticClient _elasticClient;
        private readonly ILogger<ElasticsearchRepository<T>> _logger;

        public ElasticsearchRepository(
            IElasticClient elasticClient,
            ILogger<ElasticsearchRepository<T>> logger
            )
        {
            _elasticClient = elasticClient;
            _logger = logger;
        }

        private string GetIndexName()
        {
            var indexName = typeof(T).GetCustomAttribute<ElasticsearchTypeAttribute>()?.RelationName ?? string.Empty;
            if (string.IsNullOrEmpty(indexName))
            {
                throw new ArgumentNullException($"{nameof(T)}没有配置MetaIndex属性");
            }
            return indexName;
        }

        /// <summary>
        /// 创建索引
        /// </summary>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public async Task CreateIndexAsync()
        {
            var indexName = GetIndexName();
            var existResponse = await _elasticClient.Indices.ExistsAsync(indexName);
            if (existResponse.Exists)
            {
                return;
            }
            //创建索引，自动转换类型，order为索引名称
            var response = await _elasticClient.Indices.CreateAsync(indexName, c => c.Map<T>(m => m.AutoMap()));
            if (!response.IsValid)
            {
                throw new Exception($"{indexName}索引创建失败:{response.DebugInformation}");
            }
        }

        /// <summary>
        /// 创建索引模板
        /// </summary>
        /// <param name="templateName">索引模板编号名称</param>
        /// <param name="templateAlias">别名</param>
        /// <param name="templatePattern">模板通配符表达式</param>
        /// <returns></returns>
        public async Task CreateTemplateAsync(string templateName, string templateAlias, string templatePattern)
        {
            var existResponse = await _elasticClient.Indices.TemplateV2ExistsAsync(templateName);
            if (existResponse.Exists)
            {
                return;
            }
            ITemplate selectorTemplate(TemplateDescriptor x) => x.Aliases(a => a.Alias(templateAlias)).Mappings<T>(m => m.AutoMap());
            IPutIndexTemplateV2Request selector(PutIndexTemplateV2Descriptor x) => x.IndexPatterns(templatePattern).Template(selectorTemplate);
            var templateResponse = await _elasticClient.Indices.PutTemplateV2Async(templateName, selector);
            if (!templateResponse.IsValid)
            {
                throw new Exception($"{templateName}索引模板添加失败：{templateResponse.DebugInformation}");
            }

        }

        /// <summary>
        /// 删除索引模板
        /// </summary>
        /// <param name="templateName">模板名称编码</param>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public async Task DeleteTemplateAsync(string templateName)
        {
            var existResponse = await _elasticClient.Indices.TemplateV2ExistsAsync(templateName);
            if (!existResponse.Exists)
            {
                return;
            }
            var response = await _elasticClient.Indices.DeleteTemplateV2Async(templateName);
            if (!response.IsValid)
            {
                throw new Exception($"{templateName}索引模板删除失败：{response.DebugInformation}");
            }
        }

        /// <summary>
        /// 单个对象添加
        /// </summary>
        /// <param name="data">数据对象</param>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public async Task InsertAsync(T data)
        {
            var indexName = GetIndexName();
            await InsertAsync(indexName, data);
        }

        /// <summary>
        /// 添加单个对象到指定索引
        /// </summary>
        /// <param name="indexName">索引名称</param>
        /// <param name="data">数据对象</param>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public async Task InsertAsync(string indexName, T data)
        {
            var response = await _elasticClient.IndexAsync(data, x => x.Index(indexName));
            if (!response.IsValid)
            {
                throw new Exception($"{indexName}添加失败");
            }
        }

        /// <summary>
        /// 批量添加
        /// </summary>
        /// <param name="datas"></param>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public async Task InsertAsync(IList<T> datas)
        {
            var indexName = GetIndexName();
            await InsertAsync(indexName, datas);
        }

        /// <summary>
        /// 批量添加对象集合到指定索引
        /// </summary>
        /// <param name="indexName">索引名称</param>
        /// <param name="datas"></param>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public async Task InsertAsync(string indexName, IList<T> datas)
        {
            if (!datas.Any())
            {
                return;
            }
            //分页推送-防止数据量过大接口超时
            int page = 1;
            int pageSize = 1000;
            int totalPage = (datas.Count + pageSize - 1) / pageSize;
            while (page <= totalPage)
            {
                try
                {
                    //本次推送的数据
                    var currentPushDatas = datas.Skip((page - 1) * pageSize).Take(pageSize).ToList();
                    var response = await _elasticClient.BulkAsync(b => b.Index(indexName).IndexMany(currentPushDatas));
                    if (!response.IsValid)
                    {
                        _logger?.LogInformation($"{indexName}批量添加失败：{JsonSerializer.Serialize(response)}");
                    }
                }
                catch (Exception ex)
                {
                    //记录日志
                    _logger?.LogError(ex, $"{indexName}批量添加异常page={page}：{ex}");
                }
                //下一页
                page++;
            }
        }

        public void BulkAll(IList<T> datas)
        {
            var index = GetIndexName();
            BulkAll(index, datas);
        }

        public void BulkAll(string indexName, IList<T> datas)
        {
            if (!datas.Any())
            {
                return;
            }
            var bulkAllObserver = _elasticClient.BulkAll(datas, b => b
                .Index(indexName)
                .BackOffTime("30s")//集群繁忙，报429错误码的时候，等待多久进行重试
                .BackOffRetries(2) //重试次数
                .RefreshOnCompleted() //
                .MaxDegreeOfParallelism(Environment.ProcessorCount)
                .Size(1000) //每次 bulk 请求时的文档数量。
            ).Wait(TimeSpan.FromSeconds(30), next =>
            {
                //执行索引并等待 1min，BulkAll 请求虽然时异步的，但是是一个阻塞操作。
            });
        }

        /// <summary>
        /// 单个更新
        /// </summary>
        /// <param name="data"></param>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public async Task UpdateAsync(T data)
        {
            var indexName = GetIndexName();
            await UpdateAsync(indexName, data);
        }

        /// <summary>
        /// 更新指定索引的对象
        /// </summary>
        /// <param name="indexName">索引名称</param>
        /// <param name="data"></param>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public async Task UpdateAsync(string indexName, T data)
        {
            DocumentPath<T> documentPath = new(data.Id);
            var response = await _elasticClient.UpdateAsync(documentPath, (p) => p.Doc(data).Index(indexName));
            if (!response.IsValid)
            {
                throw new Exception($"{indexName}更新失败");
            }
        }

        /// <summary>
        /// 批量更新
        /// </summary>
        /// <param name="datas"></param>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public async Task UpdatesAsnyc(IList<T> datas)
        {
            var indexName = GetIndexName();
            await UpdatesAsnyc(indexName, datas);
        }

        /// <summary>
        /// 批量更新指定索引的对象
        /// </summary>
        /// <param name="indexName">索引名称</param>
        /// <param name="datas"></param>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public async Task UpdatesAsnyc(string indexName, IList<T> datas)
        {
            var updateDescriptor = new BulkDescriptor();
            foreach (var data in datas)
            {
                updateDescriptor.Update<T>(i => i
                  .Index(indexName)
                  .Id(data.Id)
                  .Doc(data)
                  .DocAsUpsert(true));
            }
            var response = await _elasticClient.BulkAsync(updateDescriptor);
            if (!response.IsValid)
            {
                throw new Exception($"{indexName}批量更新失败");
            }
        }


        /// <summary>
        /// 单个删除
        /// </summary>
        /// <param name="id"></param>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public async Task DeleteAsync(string id)
        {
            var indexName = GetIndexName();
            await DeleteAsync(indexName, id);
        }

        /// <summary>
        /// 单个删除
        /// </summary>
        /// <param name="indexName">索引名称</param>
        /// <param name="id">数据id</param>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public async Task DeleteAsync(string indexName, string id)
        {
            if (string.IsNullOrWhiteSpace(id))
            {
                return;
            }
            var response = await _elasticClient.DeleteAsync<T>(id, i => i.Index(indexName));
            if (!response.IsValid)
            {
                throw new Exception($"{indexName}删除失败");
            }
        }

        /// <summary>
        /// 批量删除
        /// </summary>
        /// <param name="ids"></param>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public async Task DeletesAsync(IList<string> ids)
        {
            var indexName = GetIndexName();
            await DeletesAsync(indexName, ids);
        }

        /// <summary>
        /// 批量删除
        /// </summary>
        /// <param name="indexName">索引名称</param>
        /// <param name="ids"></param>
        /// <returns></returns>
        /// <exception cref="Exception"></exception>
        public async Task DeletesAsync(string indexName, IList<string> ids)
        {
            if (!ids.Any())
            {
                return;
            }
            var bulkQuest = new BulkRequest()
            {
                Operations = new List<IBulkOperation>()
            };

            foreach (var id in ids)
            {
                bulkQuest.Operations.Add(new BulkDeleteOperation<T>(id));
            }
            var response = await _elasticClient.BulkAsync(bulkQuest);

            if (!response.IsValid)
            {
                throw new Exception($"{indexName}批量删除失败");
            }
        }

        /// <summary>
        /// 根据Id获取对象
        /// </summary>
        /// <param name="id"></param>
        /// <returns></returns>
        public async Task<T> GetAsync(string id)
        {
            var indexName = GetIndexName();

            return await GetAsync(indexName, id);
        }

        /// <summary>
        /// 根据Id获取对象
        /// </summary>
        /// <param name="indexName">索引名称</param>
        /// <param name="id"></param>
        /// <returns></returns>
        public async Task<T> GetAsync(string indexName, string id)
        {
            if (string.IsNullOrWhiteSpace(id))
            {
                return null;
            }

            var response = await _elasticClient.GetAsync<T>(id, x => x.Index(indexName));

            return response.Source;
        }

        /// <summary>
        /// 根据Id集合批量获取数据集合
        /// </summary>
        /// <param name="ids"></param>
        /// <returns></returns>
        public async Task<IList<T>> GetListAsync(IList<string> ids)
        {
            var indexName = GetIndexName();

            return await GetListAsync(indexName, ids);
        }

        /// <summary>
        /// 根据Id集合批量获取数据集合
        /// </summary>
        /// <param name="indexName">索引名称</param>
        /// <param name="ids"></param>
        /// <returns></returns>
        public async Task<IList<T>> GetListAsync(string indexName, IList<string> ids)
        {
            if (!ids.Any())
            {
                return new List<T>();
            }
            var response = await _elasticClient.SearchAsync<T>(x => x.Query(o => o.Ids(m => m.Values(ids))).Index(indexName));

            return response.Documents.ToList();
        }

        /// <summary>
        /// 普通分页搜索
        /// </summary>
        /// <param name="condition"></param>
        /// <returns></returns>
        public async Task<PagedResult<T>> SearchPageAsync(int page, int limit, Func<QueryContainerDescriptor<T>, QueryContainer> queryDescriptor, Func<SortDescriptor<T>, SortDescriptor<T>> sortDescriptor = null)
        {
            var indexName = GetIndexName();

            return await SearchPageAsync(indexName, page, limit, queryDescriptor, sortDescriptor);
        }

        /// <summary>
        /// 普通分页搜索
        /// </summary>
        /// <param name="indexName">索引名称</param>
        /// <param name="condition"></param>
        /// <returns></returns>
        public async Task<PagedResult<T>> SearchPageAsync(string indexName, int page, int limit, Func<QueryContainerDescriptor<T>, QueryContainer> queryDescriptor, Func<SortDescriptor<T>, SortDescriptor<T>> sortDescriptor = null)
        {
            //sortDescriptor
            sortDescriptor = sortDescriptor ?? (sd => sd.Descending(p => p.CreatedAt));

            ISearchRequest searchSelector(SearchDescriptor<T> s) => s
                .Index(indexName)
                .Query(queryDescriptor)
                .From((page - 1) * limit)
                .Size(limit)
                .Sort(sortDescriptor);

            var response = await _elasticClient.SearchAsync<T>(searchSelector);

            ICountRequest countQuery(CountDescriptor<T> s) => s
                .Index(indexName)
                .Query(queryDescriptor);

            var countResponse = await _elasticClient.CountAsync<T>(countQuery);

            return new PagedResult<T>
            {
                Data = response.Documents.ToList(),
                PageIndex = page,
                PageSize = limit,
                TotalCount = countResponse.Count
            };
        }

        /// <summary>
        /// 滚动查询
        /// </summary>
        /// <param name="condition"></param>
        /// <returns></returns>
        public async Task<List<T>> ScrollSearchAsync(int limit, Func<QueryContainerDescriptor<T>, QueryContainer> queryDescriptor, Func<SortDescriptor<T>, SortDescriptor<T>> sortDescriptor = null)
        {
            var indexName = GetIndexName();

            return await ScrollSearchAsync(indexName, limit, queryDescriptor, sortDescriptor);
        }

        /// <summary>
        /// 滚动查询
        /// </summary>
        /// <param name="indexName">索引名称</param>
        /// <param name="condition"></param>
        /// <returns></returns>
        public async Task<List<T>> ScrollSearchAsync(string indexName, int limit, Func<QueryContainerDescriptor<T>, QueryContainer> queryDescriptor, Func<SortDescriptor<T>, SortDescriptor<T>> sortDescriptor = null)
        {
            //sortDescriptor
            sortDescriptor = sortDescriptor ?? (sd => sd.Descending(p => p.CreatedAt));

            ISearchRequest searchSelector(SearchDescriptor<T> s) => s
                .Index(indexName)
                .Query(queryDescriptor)
                .Size(limit)
                .Sort(sortDescriptor)
                .Scroll(Time.MinusOne);

            var response = await _elasticClient.SearchAsync<T>(searchSelector);

            var returnDatas = new List<T>();
            returnDatas.AddRange(response.Documents);
            string scrollId = response.ScrollId;

            while (true)
            {
                var scrollResponse = await _elasticClient.ScrollAsync<T>("1m", scrollId);
                if (!scrollResponse.Documents.Any())
                {
                    break;
                }

                returnDatas.AddRange(scrollResponse.Documents);
            }

            return returnDatas;
        }
    }
    #endregion

    #region Interface
    public interface IElasticsearchRepository<T> where T : ElasticsearchEntity
    {
        Task CreateIndexAsync();
        Task CreateTemplateAsync(string templateName, string templateAlias, string templatePattern);
        Task DeleteTemplateAsync(string templateName);
        Task InsertAsync(T data);
        Task InsertAsync(string indexName, T data);
        Task InsertAsync(IList<T> datas);
        Task InsertAsync(string indexName, IList<T> datas);
        void BulkAll(IList<T> datas);
        void BulkAll(string indexName, IList<T> datas);
        Task UpdateAsync(T data);
        Task UpdateAsync(string indexName, T data);
        Task UpdatesAsnyc(IList<T> datas);
        Task UpdatesAsnyc(string indexName, IList<T> datas);
        Task DeleteAsync(string id);
        Task DeleteAsync(string indexName, string id);
        Task DeletesAsync(IList<string> ids);
        Task DeletesAsync(string indexName, IList<string> ids);
        Task<T> GetAsync(string id);
        Task<T> GetAsync(string indexName, string id);
        Task<IList<T>> GetListAsync(IList<string> ids);
        Task<IList<T>> GetListAsync(string indexName, IList<string> ids);
        Task<PagedResult<T>> SearchPageAsync(int page, int limit, Func<QueryContainerDescriptor<T>, QueryContainer> queryDescriptor, Func<SortDescriptor<T>, SortDescriptor<T>> sortDescriptor = null);
        Task<PagedResult<T>> SearchPageAsync(string indexName, int page, int limit, Func<QueryContainerDescriptor<T>, QueryContainer> queryDescriptor, Func<SortDescriptor<T>, SortDescriptor<T>> sortDescriptor = null);
        Task<List<T>> ScrollSearchAsync(int limit, Func<QueryContainerDescriptor<T>, QueryContainer> queryDescriptor, Func<SortDescriptor<T>, SortDescriptor<T>> sortDescriptor = null);
        Task<List<T>> ScrollSearchAsync(string indexName, int limit, Func<QueryContainerDescriptor<T>, QueryContainer> queryDescriptor, Func<SortDescriptor<T>, SortDescriptor<T>> sortDescriptor = null);
    } 
    #endregion
}
